/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package cn.com.sina.erp.easymr;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.Tool;

/**
 *
 * @author guti
 */
public class MrGeneralTest extends Configured implements Tool {

    public static class MapperFavorite extends Mapper<LongWritable, Text, TextPair, Text> {
        @Override
        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String[] fields = value.toString().split(" ");
            if (fields.length < 2) {
                return;
            }
            context.write(new TextPair(fields[0], "0-favo"), new Text(fields[1]));
        }
    }

    public static class MapperDiet extends Mapper<LongWritable, Text, TextPair, Text> {
        private HashSet<String> playerList = new HashSet<String>();
        @Override
        public void setup(Context context) throws IOException {
            String path = EasyMr.getCacheFilePath("players", context);
            BufferedReader reader = new BufferedReader(new FileReader(path));
            while (true) {
                String line = reader.readLine();
                if (line == null) {
                    break;
                }
                this.playerList.add(line);
            }
        }
        @Override
        public void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            String[] fields = value.toString().split(" ");
            if (fields.length < 2) {
                return;
            }
            if (this.playerList.contains(fields[0])) {
                context.write(new TextPair(fields[0], "1-diet"), new Text(fields[1]));
            }
        }
    }

    public static class ReducerCredit extends Reducer<TextPair, Text, Text, Text> {
        
        private int numDiets = 0;

        @Override
        public void setup(Context context) {
            String numDietsString = EasyMr.getProtectedArg("numDiets", context);
            this.numDiets = Integer.parseInt(numDietsString);
        }
        @Override
        public void reduce(TextPair key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            String name = null;
            String favorite = null;
            int count = 0;
            Iterator iterator = values.iterator();
            
            while (iterator.hasNext()) {
                String v = iterator.next().toString();
                String first = key.getFirst().toString();
                String second = key.getSecond().toString();
                if (second.equals("0-favo")) {
                    name = first;
                    favorite = v;
                } else if (v.equals(favorite)) {
                    count++;
                }
            }
            if (name != null) {
                context.write(new Text(name), new Text((1.0 * count / this.numDiets) + ""));
            }
        }
    }

    public int run(String[] args) throws Exception {

        EasyMr easyMr = new EasyMr();
        easyMr.helpFooter.add("--mr=MrGeneralTest");
        easyMr.helpFooter.add("--input");
        easyMr.helpFooter.add("    /prefix/easymr_dev/favorite,MapperFavorite,lzo");
        easyMr.helpFooter.add("    /prefix/easymr_dev/diet/{2011/05/25-2011/05/31},MapperDiet");
        easyMr.helpFooter.add("--output");
        easyMr.helpFooter.add("    /prefix/easymr_dev/output");
        easyMr.helpFooter.add("--cache");
        easyMr.helpFooter.add("    /prefix/easymr_dev/cache/players,players");
        easyMr.helpFooter.add("--protectedargs");
        easyMr.helpFooter.add("    numDiets=7");
        easyMr.helpFooter.add("--numreduces=2");
        easyMr.helpFooter.add("--force");
        easyMr.helpFooter.add("Author: someone@somewhere");
        easyMr.registerClass(MapperFavorite.class);
        easyMr.registerClass(MapperDiet.class);
        easyMr.registerClass(ReducerCredit.class);
        easyMr.mapOutputKeyType = "textpair";    // uncomment this line to use textpair
        easyMr.takeArgs(args);
        easyMr.run(MrGeneralTest.class);
        return 0;
    }
}
